package com.learn.kafka.partition;

import com.learn.kafka.config.ProducerProperties;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Random;
import java.util.concurrent.ExecutionException;

/**
 * 自定义分区选择的生产者（继承通用配置父类）
 * @author: learn.kafka
 */
public class ProducerHandPartition extends ProducerProperties {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        new ProducerHandPartition().sendMessages();
    }

    // 构造方法：初始化当前生产者的特殊配置
    public ProducerHandPartition() {
        super(); // 调用父类构造方法，加载通用配置
        initProducerConfig();
    }

    // 初始化当前生产者的特殊配置（覆盖或补充父类配置）
    private void initProducerConfig() {
        // 覆盖acks策略为"-1"（即all，父类默认是"1"）
        setProperty(ProducerConfig.ACKS_CONFIG, "-1");
        // 覆盖linger.ms为10ms（父类默认是5ms）
        setProperty(ProducerConfig.LINGER_MS_CONFIG, "10");
        // 覆盖max.block.ms为5000ms（父类默认是3000ms）
        setProperty(ProducerConfig.MAX_BLOCK_MS_CONFIG, "5000");
    }

    // 发送消息的核心逻辑：随机指定分区发送
    private void sendMessages() throws ExecutionException, InterruptedException {
        // 从父类获取生产者实例（泛型为<String, Integer>，匹配键值类型）
        Producer<Integer, String> producer = createProducer();
        String topic = "qs4part2673";

        // 获取主题的分区数量
        int partitionSize = producer.partitionsFor(topic).size();
        System.out.println("Partition size: " + partitionSize);

        // 发送10条消息，随机指定分区
        for (int i = 0; i < 10; i++) {
            // 随机选择分区（0到partitionSize-1之间）
            int partition = new Random().nextInt(partitionSize);
            ProducerRecord<Integer, String> producerRecord =
                    new ProducerRecord<>(topic, partition, i, i+"");

            // 同步发送并获取元数据
            RecordMetadata metadata = producer.send(producerRecord).get();
            System.out.printf("发送到分区: %d, 偏移量: %d, 消息值: %d%n",
                    metadata.partition(), metadata.offset(), i);
        }

        // 关闭生产者
        producer.close();
    }
}
